package com.hudson.codec.domain;

import com.hudson.codec.hudson.utils.CodecUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.Unpooled;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class MessageExt {

    private String topic;              // 消息主题
    private int queueId;               // 队列 ID
    private String tags;
    private long bornTimestamp;        // 消息创建时间
    private long storeTimestamp;       // 消息存储时间
    private long commitLogOffset;      // 消息在 CommitLog 中的偏移量
    private int reconsumeTimes;        // 消息重试次数
    private byte[] body;               // 消息体

    public MessageExt(String topic, int queueId,String tags, long bornTimestamp, long storeTimestamp, long commitLogOffset, int reconsumeTimes, byte[] body) {
        this.topic = topic;
        this.queueId = queueId;
        this.tags = tags;
        this.bornTimestamp = bornTimestamp;
        this.storeTimestamp = storeTimestamp;
        this.commitLogOffset = commitLogOffset;
        this.reconsumeTimes = reconsumeTimes;
        this.body = body;
    }

    public MessageExt() {
    }

    public static Builder builder(){
        return new Builder();
    }

    // Builder 静态内部类
    public static class Builder {
        private String topic;
        private int queueId;
        private String tags;
        private long bornTimestamp;
        private long storeTimestamp;
        private long commitLogOffset;
        private int reconsumeTimes;
        private byte[] body;

        public Builder topic(String topic) {
            this.topic = topic;
            return this;
        }

        public Builder queueId(int queueId) {
            this.queueId = queueId;
            return this;
        }

        public Builder tag(String tags) {
            this.tags = tags;
            return this;
        }

        public Builder bornTimestamp(long bornTimestamp) {
            this.bornTimestamp = bornTimestamp;
            return this;
        }

        public Builder storeTimestamp(long storeTimestamp) {
            this.storeTimestamp = storeTimestamp;
            return this;
        }

        public Builder commitLogOffset(long commitLogOffset) {
            this.commitLogOffset = commitLogOffset;
            return this;
        }

        public Builder reconsumeTimes(int reconsumeTimes) {
            this.reconsumeTimes = reconsumeTimes;
            return this;
        }

        public Builder body(byte[] body) {
            this.body = body;
            return this;
        }

        public MessageExt build() {
            return new MessageExt(topic, queueId,tags, bornTimestamp, storeTimestamp, commitLogOffset, reconsumeTimes, body);
        }
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public int getQueueId() {
        return queueId;
    }

    public void setQueueId(int queueId) {
        this.queueId = queueId;
    }

    public long getBornTimestamp() {
        return bornTimestamp;
    }

    public void setBornTimestamp(long bornTimestamp) {
        this.bornTimestamp = bornTimestamp;
    }

    public long getStoreTimestamp() {
        return storeTimestamp;
    }

    public void setStoreTimestamp(long storeTimestamp) {
        this.storeTimestamp = storeTimestamp;
    }

    public long getCommitLogOffset() {
        return commitLogOffset;
    }

    public void setCommitLogOffset(long commitLogOffset) {
        this.commitLogOffset = commitLogOffset;
    }

    public int getReconsumeTimes() {
        return reconsumeTimes;
    }

    public void setReconsumeTimes(int reconsumeTimes) {
        this.reconsumeTimes = reconsumeTimes;
    }

    public byte[] getBody() {
        return body;
    }

    public void setBody(byte[] body) {
        this.body = body;
    }

    public String getTags() {
        return tags;
    }

    public void setTags(String tags) {
        this.tags = tags;
    }

    @Override
    public String toString() {
        return "MessageExt{" +
                "topic='" + topic + '\'' +
                ", queueId=" + queueId +
                ", bornTimestamp=" + bornTimestamp +
                ", storeTimestamp=" + storeTimestamp +
                ", commitLogOffset=" + commitLogOffset +
                ", reconsumeTimes=" + reconsumeTimes +
                ", body=" + Arrays.toString(body) +
                '}';
    }

    /**
     * 从byte[]反序列化MessageExt
     */
    public static MessageExt deserialize(byte[] data) {
        if (data == null || data.length == 0) {
            return new MessageExt();
        }

        ByteBuf byteBuf = Unpooled.wrappedBuffer(data);
        try {
            return deserialize(byteBuf);
        } finally {
            // Unpooled.wrappedBuffer创建的ByteBuf不需要释放
            // byteBuf.release(); // 注意：wrappedBuffer不需要手动释放
        }
    }

    /**
     * 序列化MessageExt到byte[]
     */
    public static byte[] serialize(MessageExt message) {
        if (message == null) {
            return new byte[0];
        }

        // 不提前计算长度，使用默认大小的ByteBuf
        ByteBuf byteBuf = Unpooled.buffer(); // 使用默认初始容量

        try {
            serialize(message, byteBuf);
            byte[] result = new byte[byteBuf.readableBytes()];
            byteBuf.readBytes(result);
            return result;
        } finally {
            if (byteBuf.refCnt() > 0) {
                byteBuf.release();
            }
        }
    }

    /**
     * 序列化MessageExt到ByteBuf
     */
    public static void serialize(MessageExt messageExt, ByteBuf out) {
        // 序列化消息主题
        CodecUtil.writeString(out, messageExt.topic);

        // 序列化队列 ID
        out.writeInt(messageExt.queueId);

        CodecUtil.writeString(out,messageExt.tags);

        // 序列化消息创建时间
        out.writeLong(messageExt.bornTimestamp);

        // 序列化消息存储时间
        out.writeLong(messageExt.storeTimestamp);

        // 序列化消息在 CommitLog 中的偏移量
        out.writeLong(messageExt.commitLogOffset);

        // 序列化消息重试次数
        out.writeInt(messageExt.reconsumeTimes);

        // 序列化消息体
        byte[] body = messageExt.body;
        if (body == null) {
            out.writeInt(0);
        } else {
            out.writeInt(body.length);
            out.writeBytes(body);
        }
    }

    /**
     * 从ByteBuf反序列化MessageExt
     */
    public static MessageExt deserialize(ByteBuf in) {
        MessageExt messageExt = new MessageExt();

        // 反序列化消息主题
        messageExt.setTopic(CodecUtil.readString(in));

        // 反序列化队列 ID
        messageExt.setQueueId(in.readInt());

        // 反序列化消息主题
        messageExt.setTags(CodecUtil.readString(in));

        // 反序列化消息创建时间
        messageExt.setBornTimestamp(in.readLong());

        // 反序列化消息存储时间
        messageExt.setStoreTimestamp(in.readLong());

        // 反序列化消息在 CommitLog 中的偏移量
        messageExt.setCommitLogOffset(in.readLong());

        // 反序列化消息重试次数
        messageExt.setReconsumeTimes(in.readInt());

        // 反序列化消息体
        int bodyLength = in.readInt();
        if (bodyLength > 0) {
            byte[] body = new byte[bodyLength];
            in.readBytes(body);
            messageExt.setBody(body);
        }

        return messageExt;
    }

    public static byte[] list2Bytes(List<MessageExt> list) {
        int length = 0;
        ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer();
        try {
            // 预留总长度字段
            byteBuf.writeInt(0);
            int totalLength = 4; // 初始长度为4（总长度字段本身）

            for (MessageExt messageExt : list) {
                byte[] serialize = MessageExt.serialize(messageExt);
                byteBuf.writeInt(serialize.length);
                byteBuf.writeBytes(serialize);
                totalLength += 4 + serialize.length; // 4字节长度 + 数据长度
            }

            // 回填总长度（不包括总长度字段本身）
            byteBuf.setInt(0, totalLength - 4);

            // 正确地将ByteBuf转换为byte[]
            byte[] result = new byte[byteBuf.readableBytes()];
            byteBuf.readBytes(result);
            return result;
        } finally {
            if (byteBuf.refCnt() > 0) {
                byteBuf.release();
            }
        }
    }

    public static List<MessageExt> parse2List(byte[] message) {
        List<MessageExt> result = new ArrayList<>();
        if (message == null || message.length == 0) {
            return result;
        }
        ByteBuf byteBuf = Unpooled.wrappedBuffer(message);

        try {
            // 读取总大小（这个字段可能是冗余的，我们主要关注后面的对象）
            int totalSize = byteBuf.readInt();

            // 循环读取每个对象
            while (byteBuf.isReadable()) {
                // 读取单个对象的大小
                if (byteBuf.readableBytes() < 4) {
                    break; // 剩余数据不足4字节，无法读取大小字段
                }

                int objectSize = byteBuf.readInt();

                // 检查是否有足够的数据读取对象
                if (byteBuf.readableBytes() < objectSize) {
                    break; // 数据不足，跳出循环
                }

                // 读取对象的字节数组
                byte[] objectBytes = new byte[objectSize];
                byteBuf.readBytes(objectBytes);

                // 反序列化为MessageExt对象
                MessageExt messageExt = MessageExt.deserialize(objectBytes);
                result.add(messageExt);
            }
        } finally {
            if (byteBuf.refCnt() > 0) {
                byteBuf.release();
            }
        }

        return result;
    }
}
